package ex.tableapi;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.expressions.TimeIntervalUnit;

import static org.apache.flink.table.api.Expressions.$;

/**
 * kafka作为输入源，通过转换，存储到mysql
 */
public class KafkaToMysqlDemo {
    public static void main(String[] args) {
    EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
    TableEnvironment tEnv = TableEnvironment.create(settings);
    //kafkas输入
    tEnv.executeSql("CREATE TABLE transactions (" +
            "    account_id  BIGINT," +
            "    amount      BIGINT," +
            "    transaction_time TIMESTAMP(3)," +
            "    WATERMARK FOR transaction_time AS transaction_time - INTERVAL '5' SECOND" +
            ") WITH (" +
            "    'connector' = 'kafka'," +
            "    'topic'     = 'test4'," +
            "    'properties.bootstrap.servers' = '192.168.1.6:9092'," +
            "    'properties.group.id' = 'testGroup'," +
            "    'scan.startup.mode' = 'earliest-offset'," +
            "    'format'    = 'csv'" +
            ")");
    //mysql输出
    tEnv.executeSql("CREATE TABLE spend_report (" +
            "    account_id BIGINT," +
            "    log_ts     TIMESTAMP(3)," +
            "    amount     BIGINT," +
            "    PRIMARY KEY (account_id, log_ts) NOT ENFORCED" +
            ") WITH (" +
            "   'connector'  = 'jdbc'," +
            "   'url'        = 'jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=Asia/Shanghai&useSSL=false'," +
            "   'table-name' = 'spend_report'," +
            "   'driver'     = 'com.mysql.cj.jdbc.Driver'," +
            "   'username'   = 'root'," +
            "   'password'   = 'liji@2024'," +
            "   'sink.buffer-flush.max-rows' = '0'" +
            ")");

    Table source = tEnv.from("transactions");

//    //定义输出
//    String createInDDL =
//            "CREATE TABLE printOutTable (" +
//                    "  `account_id` BIGINT," +
//                    "  `log_ts` TIMESTAMP(3)," +
//                    "  `amount` BIGINT" +
//                    ") WITH (" +
//                    "  'connector' = 'print'" +
//                    ")";
//
//    tEnv.executeSql(createInDDL);

    //转换
    Table result=source.select(
            $("account_id"),
            $("transaction_time").floor(TimeIntervalUnit.HOUR).as("log_ts"),
            $("amount"))
            .groupBy($("account_id"), $("log_ts"))
            .select(
                    $("account_id"),
                    $("log_ts"),
                    $("amount").sum().as("amount"));

    result.insertInto("spend_report").execute();

    //查看转换后格式
    result.printSchema();
    //转换结果输出到控制台
//    result.executeInsert("printOutTable");
}
}
